package com.atguigu.PracticalProject.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.PracticalProject.app.BaseAppV1;
import com.atguigu.PracticalProject.bean.TradeCourseOrderBean;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.function.DimAsyncFunction;
import com.atguigu.PracticalProject.util.AtguiguUtil;
import com.atguigu.PracticalProject.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTradeCourseOrderWindow extends BaseAppV1 {

    public static void main(String[] args) {
        new DwsTradeCourseOrderWindow().init(
                4001,
                2,
                "DwsTradeCourseOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        //下单人数+下单次数+下单金额   学科+类别+课程
        //{"id":"68632","user_id":"3587","course_id":"332","course_name":"尚硅谷Nginx教程",
        // "subject_id":3,"order_id":"62135","origin_amount":"200.0","coupon_reduce":"0.0",
        // "final_amount":"200.0","create_time":"2022-09-02 14:26:01","update_time":null,
        // "province_id":"2","session_id":"81c180f6-d8bc-4550-bb8e-8748969ea6b0","order_status":"1001",
        // "ts":1662099960}

        //1。按照order_detail_id去重
        SingleOutputStreamOperator<JSONObject> distinctedStream = distinctByOrderDetailId(stream);
       // distinctedStream.print();

       //2.把数据封装到pojo中
        SingleOutputStreamOperator<TradeCourseOrderBean> beanStream = parseToPojo(distinctedStream);

//        //3.按照sku_id分组，开窗聚合
        SingleOutputStreamOperator<TradeCourseOrderBean> beanStreamWithoutDim = windowAndAgg(beanStream);


//        //4.补充维度信息
        SingleOutputStreamOperator<TradeCourseOrderBean> beanStreamWithDim = addDim(beanStreamWithoutDim);

        //5.写出放到doris中
        writeToDoris(beanStreamWithDim);


    }

    private void writeToDoris(SingleOutputStreamOperator<TradeCourseOrderBean> beanStreamWithDim) {
        beanStreamWithDim
                .map(bean->{
                    SerializeConfig conf = new SerializeConfig();
                    conf.propertyNamingStrategy= PropertyNamingStrategy.SnakeCase;
                    return JSON.toJSONString(bean,conf);
                })
                .addSink(FlinkSinkUtil.getDorisSink("DwsTradeCourseOrderWindow"));

    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> addDim(SingleOutputStreamOperator<TradeCourseOrderBean> beanStreamWithoutDim) {

        //课程维度
        SingleOutputStreamOperator<TradeCourseOrderBean> courseStream = AsyncDataStream
                .unorderedWait(
                        beanStreamWithoutDim,
                        new DimAsyncFunction<TradeCourseOrderBean>() {

                            @Override
                            public String getTable() {
                                return "dim_course_info";
                            }

                            @Override
                            public String getId(TradeCourseOrderBean input) {
                                return input.getCourseId();
                            }

                            @Override
                            public void addDim(JSONObject dim, TradeCourseOrderBean bean) {
                                bean.setCourseName(dim.getString("COURSE_NAME"));
                                bean.setSubjectId(dim.getString("SUBJECT_ID"));

                            }
                        },
                        60,
                        TimeUnit.SECONDS
                );

        //学科维度
        SingleOutputStreamOperator<TradeCourseOrderBean> subjectStream = AsyncDataStream
                .unorderedWait(
                        courseStream,
                        new DimAsyncFunction<TradeCourseOrderBean>() {
                            @Override
                            public String getTable() {
                                return "dim_base_subject_info";
                            }

                            @Override
                            public String getId(TradeCourseOrderBean input) {
                                return input.getSubjectId();
                            }

                            @Override
                            public void addDim(JSONObject dim, TradeCourseOrderBean bean) {
                                bean.setSubjectName(dim.getString("SUBJECT_NAME"));
                                bean.setCategoryId(dim.getString("CATEGORY_ID"));

                            }
                        },
                        60,
                        TimeUnit.SECONDS
                );

        //类别维度
        SingleOutputStreamOperator<TradeCourseOrderBean> categoryStream = AsyncDataStream.unorderedWait(
                subjectStream,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category_info";
                    }

                    @Override
                    public String getId(TradeCourseOrderBean input) {
                        return input.getCategoryId();
                    }

                    @Override
                    public void addDim(JSONObject dim, TradeCourseOrderBean bean) {
                        bean.setCategoryName(dim.getString("CATEGORY_NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        return categoryStream;



    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> windowAndAgg(SingleOutputStreamOperator<TradeCourseOrderBean> beanStream) {

        return
                beanStream
                .map(json ->{
                    System.out.println(json);
                    return json;
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts)-> bean.getTs())
                )
                .keyBy(TradeCourseOrderBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeCourseOrderBean>() {
                            @Override
                            public TradeCourseOrderBean reduce(TradeCourseOrderBean bean1,
                                                               TradeCourseOrderBean bean2) throws Exception {
                                bean1.setFinalAmount(bean1.getFinalAmount().add(bean2.getFinalAmount()));
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context ctx,
                                                Iterable<TradeCourseOrderBean> elements,
                                                Collector<TradeCourseOrderBean> out) throws Exception {
                                TradeCourseOrderBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(ctx.window().getStart()));

                                out.collect(bean);


                            }
                        });
    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> parseToPojo(SingleOutputStreamOperator<JSONObject> distinctedStream) {
        return
                distinctedStream
                .map(new MapFunction<JSONObject, TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean map(JSONObject value) throws Exception {
                        String amount = value.getString("final_amount");
                        BigDecimal final_amount = new BigDecimal(0);
                        if (amount != null ){
                            final_amount=new BigDecimal(amount);
                        }
                        return TradeCourseOrderBean
                                .builder()
                                .courseId(value.getString("course_id"))
                                .finalAmount(final_amount)
                                .ts(value.getLong("ts")*10000)
                                .build();
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> distinctByOrderDetailId(DataStreamSource<String> stream) {
        // stream.print();
        return stream
                .map(json -> JSON.parseObject(json))

                .keyBy(obj -> obj.getString("order_id"))
                .process(
                        new
                                KeyedProcessFunction<String, JSONObject, JSONObject>() {
                                    private ValueState<JSONObject> maxTsDateState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        //将第一条数据存储到状态中
                                        maxTsDateState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("maxTsDateState", JSONObject.class));
                                    }

                                    @Override
                                    public void processElement(JSONObject value,
                                                               Context ctx,
                                                               Collector<JSONObject> out) throws Exception {
                                        // 第一条数据来的时候，把数据存到状态中，注册5s的定时器
                                        if (maxTsDateState.value() == null) {
                                            maxTsDateState.update(value);
                                            ctx.timerService().registerProcessingTimeTimer(
                                                    ctx.timerService().currentProcessingTime() + 5000L
                                            );
                                        } else {
                                            //如果不是第一条，则与状态的额数据做对比，状态更新为最大
                                            String last = maxTsDateState.value().getString("ts");
                                            String current = value.getString("ts");

                                            // 判断时间大小
                                            if (AtguiguUtil.isLarger(current, last)) {
                                                maxTsDateState.update(value);
                                            }
                                        }
                                    }

                                    @Override
                                    public void onTimer(long timestamp,
                                                        OnTimerContext ctx,
                                                        Collector<JSONObject> out) throws Exception {
                                        // 定时器触发的时候
                                        out.collect(maxTsDateState.value());
                                    }
                                }
                );
    }

}
